package com.asiainfo.dacp.datastash.broker.leader;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.leader.LeaderSelectorListenerAdapter;
import org.apache.curator.framework.state.ConnectionState;
import org.mortbay.log.Log;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.asiainfo.dacp.datastash.broker.facade.UnitFailoverFacade;
import com.asiainfo.dacp.datastash.broker.facade.service.UnitInitService;
import com.asiainfo.dacp.datastash.broker.facade.service.UnitLoaderService;
import com.asiainfo.dacp.datastash.broker.leader.service.InterService;
import com.asiainfo.dacp.datastash.broker.leader.service.MonitorBrokerService;
import com.asiainfo.dacp.datastash.broker.listener.AbstractQueueListener;
import com.asiainfo.dacp.datastash.broker.tracelog.StashTraceLogVo;
import com.asiainfo.dacp.datastash.broker.tracelog.StashTraceLogger;
import com.asiainfo.dacp.datastash.broker.util.LeaderElection;

@Service
public class LeaderService {
	private String clzName = this.getClass().getSimpleName();

	@Autowired
	private UnitInitService stashUnitInitService;

	@Autowired
	private UnitLoaderService unitLoaderService;
	
	@Autowired
	private InterService interService;

	@Autowired
	private MonitorBrokerService monitorBrokerService;

	@Autowired
	private UnitFailoverFacade unitFailoverFacade;

	@Autowired
	private StashTraceLogger tracelog;

	private volatile boolean running = true;
	private final int interval_s = 4;

	public void start() {
		// 实现选主，由主来完成这个服务的调用

		new Thread(() -> {
			try {
				new LeaderElection(new LeaderSelectorListenerAdapter() {
					public void takeLeadership(CuratorFramework curatorFramework) throws Exception {
						// 批次更新，依赖检查

						tracelog.debug(new StashTraceLogVo(clzName, "unitInitJob:election", "选主成功"));
						
						
						//更新broker集群－－监控
						monitorBrokerService.service();
						
						//startFlowCheckService();
						
						//启动外部接口服务，接受外部对流程的指令，重做等
						startInterService();
						
						//
						doOnce();
						
						//
						doAlways();
					}

					@Override
					public void stateChanged(CuratorFramework client, ConnectionState newState) {
						Log.warn("LEADER 状态变更为: {}", newState.toString());
						if (newState == ConnectionState.LOST) {
							stop();
						}
					}
				}).startup();

			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}).start();
	}
	
	public void startFlowCheckService(){
		new Thread(new AbstractQueueListener() {
			@Override
			protected void handle() {
				try {
					unitFailoverFacade.unitFailover();
					unitFailoverFacade.taskFailover();
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
			@Override
			protected int getIntervals() {
				return interval_s;
			}
			@Override
			protected boolean isRunning() {
				return running;
			}
		}).start(); 
	}
	
	public void startInterService(){
		new Thread(new AbstractQueueListener() {
			@Override
			protected void handle() {
				try { 
					//inter service
					//前台操作接口
					interService.service();
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
			@Override
			protected int getIntervals() {
				return interval_s;
			}
			@Override
			protected boolean isRunning() {
				return running;
			}
		}).start();
	}
	
	public void doOnce(){
		//全量同步
		
	}
	
	public void doAlways(){
		while (running) {
			try {
				//创建批次
				unitLoaderService.unitLoad();
				//放入队列
				stashUnitInitService.updateWaitQueue();

			} catch (Exception e) {
				e.printStackTrace();
			}
			try {
				Thread.sleep(interval_s * 1000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

	public void stop() {
		running = false;
	}

}
